// Copyright (c) 2023 Huawei Device Co., Ltd.
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use crate::event_handler::inner_queue::InnerQueue;
use crate::event_handler::InnerEvent;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::SystemTime;
use tokio::{sync::Notify, time::timeout};

const IDLE_QUEUE_INDEX: usize = Priority::Idle.as_usize();

#[derive(Clone, Copy)]
/// Event priority.
pub enum Priority {
    /// Immediate priority.
    Immediate = 0,
    /// High priority.
    High = 1,
    /// Low priority.
    Low = 2,
    /// Idle priority.
    Idle = 3,
}

impl Priority {
    /// Gets priority number.
    pub const fn as_usize(&self) -> usize {
        *self as usize
    }
}

pub(crate) struct EventQueue<A> {
    finished: AtomicBool,
    inner: Mutex<Inner<A>>,
    notify: Arc<Notify>,
}

struct Inner<A> {
    idle: bool,
    idle_time: SystemTime,
    wake_up: SystemTime,
    queue: [InnerQueue<A>; IDLE_QUEUE_INDEX + 1],
}

impl<A> Inner<A> {
    fn new() -> Self {
        Inner {
            idle: true,
            idle_time: SystemTime::UNIX_EPOCH,
            wake_up: SystemTime::UNIX_EPOCH,
            queue: [
                InnerQueue::new(),
                InnerQueue::new(),
                InnerQueue::new(),
                InnerQueue::new(),
            ],
        }
    }

    fn need_notify(&self, event: &InnerEvent<A>, priority: Priority) -> bool {
        match priority {
            Priority::Idle => false,
            _ => self.wake_up == SystemTime::UNIX_EPOCH || self.wake_up >= event.handle_time(),
        }
    }

    fn put(&mut self, event: InnerEvent<A>, priority: Priority) {
        let queue = &mut self.queue[priority.as_usize()];
        queue.put(event);
    }

    fn pick_non_idle(&mut self) -> Option<InnerEvent<A>> {
        let mut index = IDLE_QUEUE_INDEX;
        for i in 0..IDLE_QUEUE_INDEX {
            if !self.queue[i].has_expired_event(&mut self.wake_up) {
                continue;
            }
            index = i;
            break;
        }

        for i in 0..index {
            self.queue[i].set_handled(0);
        }

        if index != IDLE_QUEUE_INDEX {
            return self.queue[index].get();
        }
        None
    }

    fn pick_idle(&mut self) -> Option<InnerEvent<A>> {
        if let Some(front) = self.queue[IDLE_QUEUE_INDEX].front() {
            if front.send_time() <= self.idle_time && front.handle_time() <= SystemTime::now() {
                return self.queue[IDLE_QUEUE_INDEX].pop_front();
            }
        }
        None
    }

    fn pick_event(&mut self) -> Option<InnerEvent<A>> {
        if let Some(event) = self.pick_non_idle() {
            self.idle = false;
            return Some(event);
        }

        if !self.idle {
            self.idle = true;
            self.idle_time = SystemTime::now();
        }

        if let Some(event) = self.pick_idle() {
            return Some(event);
        }
        None
    }
}

impl<A> EventQueue<A> {
    pub(crate) fn new() -> EventQueue<A> {
        EventQueue {
            finished: AtomicBool::new(false),
            inner: Mutex::new(Inner::new()),
            notify: Arc::new(Notify::new()),
        }
    }

    pub(crate) fn put(&self, event: InnerEvent<A>, priority: Priority) {
        let mut inner = self.inner.lock().unwrap();
        let need_notify = inner.need_notify(&event, priority);
        inner.put(event, priority);
        drop(inner);
        if need_notify {
            self.notify.notify_one();
        }
    }

    pub(crate) fn finish(&self) {
        self.finished.store(true, Ordering::Relaxed);
        self.notify.notify_one();
    }

    pub(crate) async fn get(&self) -> Option<InnerEvent<A>> {
        while !self.finished.load(Ordering::Relaxed) {
            let wake_up = {
                let mut inner = self.inner.lock().unwrap();
                inner.wake_up = SystemTime::UNIX_EPOCH;
                if let Some(e) = inner.pick_event() {
                    return Some(e);
                }
                inner.wake_up
            };

            if let Ok(duration) = wake_up.duration_since(SystemTime::now()) {
                let _ = timeout(duration, self.notify.notified()).await;
            } else if wake_up == SystemTime::UNIX_EPOCH {
                self.notify.notified().await;
            }
        }
        None
    }
}
